package com.zws.cucumber.bootstrap.cass;

import com.zws.cucumber.ZwsUtils;
import com.zws.cucumber.util.cass.ZwsCassContext;
import org.apache.commons.io.FileUtils;
import org.springframework.core.io.Resource;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;

import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.file.Files;
import java.nio.file.StandardCopyOption;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

import static com.zws.cucumber.bootstrap.cass.ZwsBootstrapUtil.*;

public final class ZwsStandaloneCassJenkinsBootstrap extends ZwsAbstractCassBootstrap {

    public static final String PATTERN = "classpath*:bootstrap-scripts/cassandra/%s";
    public static final String PREPARE_ENV_FILE = "prepareEnv.sh";
    public static final String PREPARE_CASSANDRA_FILE = "prepareCassandra.sh";
    public static final String KILL_CASSANDRA_FILE = "killCassandra.sh";
    public static final String PATTERN_PREPARE_ENV = PATTERN.formatted(PREPARE_ENV_FILE);
    public static final String PATTERN_PREPARE_CASSANDRA = PATTERN.formatted(PREPARE_CASSANDRA_FILE);
    public static final String PATTERN_KILL_CASSANDRA = PATTERN.formatted(KILL_CASSANDRA_FILE);

    private static volatile boolean isInit = false;

    private int cassandraPort = 0;
    private int backdoorPort = 0;
    private String parentDirectory = null;
    private String directory = null;

    // fast load
    private String creationScriptPath;
    private String dataScriptPath;
    private String keyspace;

    public ZwsStandaloneCassJenkinsBootstrap() {
        cassandraPort = ZwsUtils.nextPort("cassandra");
        backdoorPort = ZwsUtils.nextPort("cassandra backdoor");
        parentDirectory = System.getProperty("user.dir") + "/zws-test-cassandra3";
    }

    @Override
    protected void prepareAndStartCassandra0() throws Exception {
        mkdir();
        PathMatchingResourcePatternResolver resolver = new PathMatchingResourcePatternResolver();

        prepareCassandra(resolver);
        startCassandra(resolver);
    }

    @Override
    public int getCassandraPort() {
        return cassandraPort;
    }

    @Override
    public int getBackdoorPort() {
        return backdoorPort;
    }

    @Override
    public String getPathOfSystemLogs() {
        return parentDirectory + "/" + directory + "/logs/system.log";
    }

    @Override
    public ZwsCassContext boot(String creationScriptPath, String dataScriptPath, String keyspace) throws Exception {
        if (!isInit) {
            isInit = true;
            this.creationScriptPath = creationScriptPath;
            this.dataScriptPath = dataScriptPath;
            this.keyspace = keyspace;
            return super.boot(creationScriptPath, dataScriptPath, keyspace);
        } else {
            ZwsBootstrapUtil.printf("standalone cassandra have been boot");
            return new ZwsCassContext();
        }

    }

    private void mkdir() {
        try {
            FileUtils.forceMkdir(FileUtils.getFile(parentDirectory));
        } catch (IOException ex) {
            throw new RuntimeException(ex);
        }
    }

    private String readDirectory(PathMatchingResourcePatternResolver resolver) throws IOException {
        Resource res = resolver.getResources(PATTERN_PREPARE_ENV)[0];
        try (BufferedReader reader = new BufferedReader(new InputStreamReader(res.getInputStream()))) {
            String line = null;
            while ((line = reader.readLine()) != null) {
                if (line.startsWith("export STANDALONE_CASSANDRA_DIR=")) {
                    return line.substring("export STANDALONE_CASSANDRA_DIR=".length());
                }
            }
        }
        return null;
    }

    private void prepareCassandra(PathMatchingResourcePatternResolver resolver) throws IOException, InterruptedException {
        ZwsBootstrapUtil.extractFile(resolver, PATTERN_PREPARE_ENV, parentDirectory + "/" + PREPARE_ENV_FILE);
        ZwsBootstrapUtil.extractFile(resolver, PATTERN_PREPARE_CASSANDRA, parentDirectory + "/" + PREPARE_CASSANDRA_FILE);
        ZwsBootstrapUtil.extractFile(resolver, PATTERN_KILL_CASSANDRA, parentDirectory + "/" + KILL_CASSANDRA_FILE);

        new File(parentDirectory + "/" + PREPARE_ENV_FILE).setExecutable(true, false);
        new File(parentDirectory + "/" + PREPARE_CASSANDRA_FILE).setExecutable(true, false);

        String[] envs = getPrepareCassandraEnvsBuilder().toArray();
        Process p1 = Runtime.getRuntime().exec(new String[] { "/bin/bash", PREPARE_CASSANDRA_FILE, "-f" }, envs, new File(parentDirectory));
        p1.getInputStream().transferTo(System.out);
        p1.getErrorStream().transferTo(System.err);
        int exitCode = p1.waitFor();
        System.out.println("exitCode: " + exitCode);
        if (exitCode != 0) {
            throw new RuntimeException("Failed to prepare Cassandra, exitCode: %d".formatted(exitCode));
        }
    }

    private void startCassandra(PathMatchingResourcePatternResolver resolver) throws IOException, InterruptedException {
        startTimer();
        directory = readDirectory(resolver);

        FileUtils.deleteQuietly(new File(parentDirectory + "/" + directory + "/logs"));
        FileUtils.deleteQuietly(new File(parentDirectory + "/" + directory + "/data"));
        FileUtils.deleteQuietly(new File(parentDirectory + "/" + directory + "/conf/cassandra-env.sh")); // not to use the default cassandra-env.sh
        FileUtils.deleteQuietly(new File(parentDirectory + "/creation.cql"));
        FileUtils.deleteQuietly(new File(parentDirectory + "/data.cql"));
        Files.copy(
                Objects.requireNonNull(this.getClass().getResourceAsStream("/" + creationScriptPath), creationScriptPath),
                new File(parentDirectory + "/creation.cql").toPath(),
                StandardCopyOption.REPLACE_EXISTING);
        Files.copy(
                Objects.requireNonNull(this.getClass().getResourceAsStream("/" + dataScriptPath), dataScriptPath),
                new File(parentDirectory + "/data.cql").toPath(),
                StandardCopyOption.REPLACE_EXISTING);

        String[] envs = getStartCassandraEnvsBuilder()
                .withEnv("cassandra_parms", String.join(
                        " ",
                        "-Dcassandra.native_transport_port=%d".formatted(getCassandraPort()),
                        "-Dbackdoor.port=%d".formatted(getBackdoorPort()),
                        "-Dcassandra-foreground=yes",
                        "-Dcassandra.unsafesystem=true",
                        "-Dcassandra.native.epoll.enabled=false",
                        "-Dcassandra.custom_query_handler_class=zws.common.TestQueryHandler",
                        "-Dcassandra.storage_port=" + ZwsUtils.nextPort("cassandra storage")))
                .withEnv("JVM_OPTS", String.join(
                        " ",
                        "-XX:+UseThreadPriorities",
                        "-XX:ThreadPriorityPolicy=42",
                        "-Xss256k",
                        "-XX:StringTableSize=1000003",
                        "-XX:+AlwaysPreTouch",
                        "-XX:-UseBiasedLocking",
                        "-XX:+UseTLAB",
                        "-XX:+ResizeTLAB",
                        "-XX:+UseNUMA",
                        "-XX:+PerfDisableSharedMem",
                        "-Djava.net.preferIPv4Stack=true",
                        "-Xms128M",
                        "-Xmx128M",
                        "-Xmn64M",
                        "-XX:ReservedCodeCacheSize=64M",
                        "-XX:-UseCompressedOops",
                        "-XX:CompileCommandFile=" + parentDirectory + "/" + directory + "/conf/hotspot_compiler"))
                .toArray();

        Process proc = Runtime.getRuntime().exec(
                new String[] { "/bin/bash", "cassandra" },
                envs,
                new File(parentDirectory + "/" + directory + "/" + "bin"));
        handleStdoutAndStderr(proc);

        String pid = waitAndGetPid();

        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            try {
                Process killProc = Runtime.getRuntime().exec(
                        new String[] { "/bin/bash", KILL_CASSANDRA_FILE, pid },
                        getPrepareCassandraEnvsBuilder().toArray(),
                        new File(parentDirectory)

                );
                killProc.getInputStream().transferTo(System.out);
                killProc.getErrorStream().transferTo(System.err);
                killProc.waitFor(120, TimeUnit.SECONDS);
                println("kill process done, pid: " + pid);
            } catch (IOException | InterruptedException e) {
                println("kill process failed, pid: " + pid);
                e.printStackTrace();
            }
        }));
        println("standalone cassandra: started, elapsed: %dms".formatted(elapsed()));

        final String keyspacePattern = "CREATE KEYSPACE IF NOT EXISTS %s WITH replication={'class' : 'SimpleStrategy', 'replication_factor':1} AND durable_writes = false";
        Process keyspaceProc = Runtime.getRuntime().exec(
                new String[] { "/bin/bash", "cqlsh", "-e", keyspacePattern.formatted(keyspace), "127.0.0.1", Integer.toString(cassandraPort) },
                envs,
                new File(parentDirectory + "/" + directory + "/" + "bin"));
        keyspaceProc.getInputStream().transferTo(System.out);
        keyspaceProc.getErrorStream().transferTo(System.err);
        keyspaceProc.waitFor(60, TimeUnit.SECONDS);
        println("standalone cassandra: create keyspace, elapsed: %dms".formatted(elapsed()));

        Process creationProc = Runtime.getRuntime().exec(
                new String[] { "/bin/bash", "cqlsh", "-k", keyspace, "-f", parentDirectory + "/" + "creation.cql", "127.0.0.1", Integer.toString(cassandraPort) },
                envs,
                new File(parentDirectory + "/" + directory + "/" + "bin"));
        creationProc.getInputStream().transferTo(System.out);
        creationProc.getErrorStream().transferTo(System.err);
        creationProc.waitFor(60, TimeUnit.SECONDS);
        println("standalone cassandra: create table, elapsed: %dms".formatted(elapsed()));

        Process loadProc = Runtime.getRuntime().exec(
                new String[] { "/bin/bash", "cqlsh", "-k", keyspace, "-f", parentDirectory + "/" + "data.cql", "127.0.0.1", Integer.toString(cassandraPort) },
                envs,
                new File(parentDirectory + "/" + directory + "/" + "bin"));
        loadProc.getInputStream().transferTo(System.out);
        loadProc.getErrorStream().transferTo(System.err);
        loadProc.waitFor(60, TimeUnit.SECONDS);
        println("standalone cassandra: load data, elapsed: %dms".formatted(elapsed()));
    }
}
